package com.csw.flink.cep

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo02Cep {
  def main(args: Array[String]): Unit = {

    /*
    001,0.001
    001,0.001
    001,0.001
    001,0.001
    001,0.002
    001,100000

    002,0.001
    002,0.001
    002,0.001
    002,0.001
    002,0.002
    002,100000

    */

    /**
      * 如果信用卡先被多次刷了一个很小的金额
      * 接着刷了一个很大金额
      * 在一段时间内
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val bsEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    val ds: DataStream[String] = env.socketTextStream("master", 8888)

    val mapDS: DataStream[(String, Double)] = ds.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1).toDouble)
    })

    val pattern: Pattern[(String, Double), (String, Double)] = Pattern
      .begin[(String, Double)]("begin", AfterMatchSkipStrategy.skipPastLastEvent()) //匹配开始
      .where(kv => kv._2 < 0.01) //条件
      .oneOrMore //循环多次:一到多次
      .next("next") //匹配下一个
      .where(kv => kv._2 > 100) //条件
      .within(Time.seconds(10)) //在10秒内完成

    //匹配数据
    val patternDS: PatternStream[(String, Double)] = CEP.pattern(mapDS.keyBy(_._1), pattern)

    //取出匹配成功的数据
    val resultDS: DataStream[(String, Double)] = patternDS.select(map => {
      val iter: Iterable[(String, Double)] = map.getOrElse("next", null)

      //大额的消费
      val last: (String, Double) = iter.last

      last
    })

    //将DataStream转换成表，指定列名
    val table: Table = bsEnv.fromDataStream(resultDS,$"id",$"pic")

    //注册成临时视图
    bsEnv.createTemporaryView("ds_table",table)

    bsEnv.executeSql(
      """
        |
        |CREATE TABLE pic_sum (
        |  id STRING,
        |  pic DOUBLE,
        |  PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/csw?useUnicode=true&characterEncoding=utf-8&useSSL=false',
        |   'table-name' = 'pic_sum',
        |   'username' = 'root',
        |   'password'= '123456'
        |)
        |
      """.stripMargin)

    bsEnv.executeSql(
      """
        |insert into pic_sum
        |select * from ds_table
        |
      """.stripMargin)
  }
}
